package com.yixun.service;

import com.alibaba.fastjson.JSONObject;
import com.google.gson.Gson;
import com.google.gson.reflect.TypeToken;
import com.yixun.component.RedisService;
import com.yixun.constant.SystemConstant;
import com.yixun.model.BreakdownInfo;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.util.ObjectUtils;
import org.springframework.util.StringUtils;

import java.lang.reflect.Type;
import java.math.BigDecimal;
import java.util.*;
import java.util.stream.Collectors;

@Service
@Slf4j
public class HeatUpService {

    @Autowired
    RedisService redisService;


    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;
    private static Gson gson = new Gson();

    private static BigDecimal heatUp = new BigDecimal("30");

    /**
     * @Desc: 站点升温过快故障--异步分析
     **/
    @Async("asyncPool")
    public void siteHeatUp(String siteNo){
        //获取站点最新2条数据
        String key = String.format(SystemConstant.SITE_DATA_KEY,siteNo);
        if(!redisService.exists(key) || redisService.llen(key)<=2){
            log.error("升温故障::站点缓存数据不存在...");
            return;
        }

        List<String> siteDataJsons = redisService.lrange(key, 0, 1);

        Type type = new TypeToken<List<String>>() {
        }.getType();



        //获取到连续两条数据
        List<String> currentSiteData = gson.fromJson(siteDataJsons.get(0), type);
        List<String> lastSiteData = gson.fromJson(siteDataJsons.get(1), type);

        System.out.println("redis订阅breakdown主题，站点升温过快故障currentSiteData="+currentSiteData.toString());
        System.out.println("redis订阅breakdown主题，站点升温过快故障lastSiteData="+lastSiteData.toString());

        Collections.replaceAll(currentSiteData, null, "0");
        Collections.replaceAll(lastSiteData, null, "0");


        //去掉两条数据中的站点和时间
        List<BigDecimal> currentSiteTemp = currentSiteData.subList(2,currentSiteData.size())
                .stream().map(BigDecimal::new).collect(Collectors.toList());
        List<BigDecimal> lastSiteTemp = lastSiteData.subList(2,lastSiteData.size())
                .stream().map(BigDecimal::new).collect(Collectors.toList());

        //遍历对比获取温差大于30的通道以及温差
        Map<Integer,BigDecimal> contr = new HashMap<>();

        for (int index = 0; index < currentSiteTemp.size(); index++) {
            BigDecimal currentTemp = currentSiteTemp.get(index);
            if(index<lastSiteData.size()){
                BigDecimal lastTemp = lastSiteTemp.get(index);
                //判断温差是否大于30
                BigDecimal difference = currentTemp.subtract(lastTemp).abs();
                if(difference.compareTo(heatUp)> -1){//温差跟30相比较
                    //根据index算出当前属于第几通道（20个测温点为一个通道）
                    Integer aisle = (index/20)+1;
                    if(ObjectUtils.isEmpty(contr.get(aisle))){//contr如果是空的
                        contr.put(aisle,difference);
                        continue;
                    }
                    if(contr.get(aisle).compareTo(difference) == -1){//不是空的，就判断是否比计算的温差大，大就更新
                        contr.put(aisle,difference);
                    }
                }
            }
        }

        //将温差过大的站点通道推送至kafka，缓存至redis
        contr.forEach((aisle,deTemp)->{

            Date currentDate = new Date();

            //是否存在redis
            //存在，更新
            //不存在，新建
            String breakdownKey = String.format(SystemConstant.BREAKDOWN_HEAT_UP_KEY, siteNo, aisle);
            BreakdownInfo breakdownInfo = null;
            if(redisService.exists(breakdownKey)){
                //把redis中的付赋给breakdownInfo
                breakdownInfo = JSONObject.parseObject(redisService.get(breakdownKey), BreakdownInfo.class);
                breakdownInfo.setRemark("温差:"+String.valueOf(deTemp));
            }else{
                breakdownInfo  = new BreakdownInfo();
                breakdownInfo.setUid(String.valueOf(redisService.incr(SystemConstant.BREAKDOWN_UID)));
                breakdownInfo.setRemark("温差:"+String.valueOf(deTemp));
                breakdownInfo.setAisleNo(String.valueOf(aisle));
                breakdownInfo.setSiteNo(siteNo);
                breakdownInfo.setStatus(BreakdownInfo.STATUS_WARNING);
                breakdownInfo.setTime(String.valueOf(currentDate.getTime()));
                breakdownInfo.setType(BreakdownInfo.TYPE_HEAT_UP);
            }

            //更新至redis,推送至kafka
            String breakdownJson = JSONObject.toJSONString(breakdownInfo);
            redisService.set(breakdownKey, breakdownJson);//"breakdown:%s_%s_HEAT_UP";
            kafkaTemplate.send(SystemConstant.BREAKDOWN_TOPIC, breakdownJson);//breakdown_data

            log.info("升温故障推送:"+siteNo+"::"+aisle+":::"+SystemConstant.dateFormat.format(currentDate));

        });

    }


}
